/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.xhy.sniffer.datacarrier.consumer;


import org.xhy.sniffer.datacarrier.EnvUtil;
import org.xhy.sniffer.datacarrier.buffer.Channels;

import java.util.ArrayList;
import java.util.List;

/**
 * BulkConsumePool 用于从多个通道（即多个 DataCarrier 实例）中消费数据，并且使用多个 {@link MultipleChannelsConsumer} 进行消费。
 * <p>
 * 在典型场景下，{@link MultipleChannelsConsumer} 的数量应该少于通道数量。
 */
public class BulkConsumePool implements ConsumerPool {
    private List<MultipleChannelsConsumer> allConsumers; // 所有的消费者线程
    private volatile boolean isStarted = false; // 标志消费者池是否已启动

    // 构造函数，根据指定的名称、线程池大小和消费周期创建消费者池
    public BulkConsumePool(String name, int size, long consumeCycle) {
        // 从环境变量中获取线程池大小配置
        size = EnvUtil.getInt(name + "_THREAD", size);
        allConsumers = new ArrayList<MultipleChannelsConsumer>(size);
        // 创建并初始化所有消费者
        for (int i = 0; i < size; i++) {
            MultipleChannelsConsumer multipleChannelsConsumer = new MultipleChannelsConsumer(
                    "DataCarrier." + name + ".BulkConsumePool." + i + ".Thread", consumeCycle);
            multipleChannelsConsumer.setDaemon(true); // 设置为守护线程
            allConsumers.add(multipleChannelsConsumer); // 添加到消费者列表
        }
    }

    /**
     * 向消费者池添加新的通道及其消费者。
     * 消费者池会自动选择负载最小的消费者线程分配该通道。
     */
    @Override
    public synchronized void add(String name, Channels channels, IConsumer consumer) {
        // 获取负载最小的消费者线程
        MultipleChannelsConsumer multipleChannelsConsumer = getLowestPayload();
        // 将新通道及其消费者分配给该线程
        multipleChannelsConsumer.addNewTarget(channels, consumer);
    }

    /**
     * 获取当前负载最小的消费者线程。
     *
     * @return 负载最小的消费者线程
     */
    private MultipleChannelsConsumer getLowestPayload() {
        MultipleChannelsConsumer winner = allConsumers.get(0); // 假设第一个消费者负载最小
        // 遍历所有消费者线程，找到负载最小的
        for (int i = 1; i < allConsumers.size(); i++) {
            MultipleChannelsConsumer option = allConsumers.get(i);
            // 比较线程的负载大小，选择更小的
            if (option.size() < winner.size()) {
                winner = option;
            }
        }
        return winner;
    }

    /**
     * 检查消费者池是否已启动。
     */
    @Override
    public boolean isRunning(Channels channels) {
        return isStarted; // 返回是否启动标志
    }

    /**
     * 关闭消费者池中所有消费者线程。
     */
    @Override
    public void close(Channels channels) {
        // 遍历所有消费者线程并关闭它们
        for (MultipleChannelsConsumer consumer : allConsumers) {
            consumer.shutdown();
        }
    }

    /**
     * 启动消费者池中所有的消费者线程。
     */
    @Override
    public void begin(Channels channels) {
        if (isStarted) { // 如果已经启动，则直接返回
            return;
        }
        // 遍历所有消费者线程并启动它们
        for (MultipleChannelsConsumer consumer : allConsumers) {
            consumer.start();
        }
        isStarted = true; // 设置启动标志
    }
}

